package net.wlm.realtime.new_job;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.wlm.realtime.bean.VehicleData;
import net.wlm.realtime.utils.JdbcUtil;
import net.wlm.realtime.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import java.time.Duration;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
/***
 * 车辆实时状态监控
 * 5分钟窗口
 */
public class RealtimeVehicleMonitor {
    public static void main(String[] args) throws Exception {
        // 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 数据源
        DataStream<String> vehicleData = KafkaUtil.consumerKafka(env, "vehicle-data");
        // 数据处理
        DataStream<VehicleData> vehicleStream = cleanData(vehicleData);
        // vehicleStream.print();
        // 数据转换
        DataStream<String> stringDataStream = handle(vehicleStream);
        // stringDataStream.print();
        // 数据输出
        String[] columns =  new String[]{"vin","timestamp","longitude", "latitude", "speed", "mileage",
                "batteryLevel", "batteryTemp", "motorTemp", "chargingStatus", "energyConsumption"};
        String updateSql = "insert into table vehicle_real.vehicle_car_info(vin,timestamp,longitude, latitude, speed, mileage,\n" +
                "batteryLevel, batteryTemp, motorTemp, chargingStatus, energyConsumption)\n" +
                "values (?, ?, ?, ?, ?, ?,?, ?,?, ?, ?)";
        JdbcUtil.sinkToClickhouseUpsertJson(stringDataStream,columns,updateSql);
        // 执行
        env.execute();
    }
    /**
     * 过滤数据转换实体类
     * @param vehicleData
     * @return
     */
    private static DataStream<VehicleData> cleanData(DataStream<String> vehicleData) {
        SingleOutputStreamOperator<VehicleData> dataStream = vehicleData.filter(json -> {
                    try {
                        return json != null && JSON.isValid(json);
                    } catch (Exception e) {
                        return false;
                    }
                })
                // 将合法JSON字符串转换为VehicleData对象
                .map(json -> JSON.parseObject(json, VehicleData.class));
        return dataStream;
    }
    private static DataStream<String> handle(DataStream<VehicleData> vehicleData) {
        // 设置水位线
        SingleOutputStreamOperator<VehicleData> streamOperator = vehicleData.assignTimestampsAndWatermarks(WatermarkStrategy.<VehicleData>forBoundedOutOfOrderness(Duration.ofMinutes(0))
                .withTimestampAssigner((element, recordTimestamp) -> element.getTimestamp()));
        // 创建窗口流
        AllWindowedStream<VehicleData, TimeWindow> vehicleWindowAllStream = streamOperator.
                windowAll(TumblingEventTimeWindows.of(Time.minutes(1)));
        // 开窗
        SingleOutputStreamOperator<String> process = vehicleWindowAllStream.
                process(new ProcessAllWindowFunction<VehicleData, String, TimeWindow>() {
            @Override
            public void process(Context ctx, Iterable<VehicleData> iter, Collector<String> out) throws Exception {
                // 3秒
                long samplingInterval = 3000;
                // 窗口开始时间
                long windowStart = ctx.window().getStart();
                // 根据键值进行自然排序，根据时间顺序进行
                TreeMap<Long, VehicleData> timeSorted = new TreeMap<>();
                iter.forEach(data -> timeSorted.put(data.getTimestamp(), data));
                Set<Long> processedTimestamps = new HashSet<>();
                // 每3秒取一条数据
                for (long ts = windowStart; ts < windowStart + 60000; ts += samplingInterval) {
                    // 获取小于等于采样点的最近数据
                    Map.Entry<Long, VehicleData> sampled = timeSorted.floorEntry(ts);
                    if (sampled != null && !processedTimestamps.contains(sampled.getKey()) && sampled.getKey() >= windowStart) {
                        JSONObject jsonObject = new JSONObject();
                        VehicleData data = sampled.getValue();
                        jsonObject.put("vin",data.getVin());
                        jsonObject.put("timestamp",data.getTimestamp());
                        jsonObject.put("longitude",data.getLongitude());
                        jsonObject.put("latitude",data.getLatitude());
                        jsonObject.put("speed",data.getSpeed());
                        jsonObject.put("mileage",data.getMileage());
                        jsonObject.put("batteryLevel",data.getBatteryLevel());
                        jsonObject.put("batteryTemp",data.getBatteryTemp());
                        jsonObject.put("motorTemp",data.getMotorTemp());
                        jsonObject.put("chargingStatus",data.getChargingStatus());
                        jsonObject.put("energyConsumption",data.getEnergyConsumption());
                        out.collect(jsonObject.toJSONString());
                        processedTimestamps.add(sampled.getKey()); // 记录已处理的时间戳
                    }
                }
            }
        });
        return process;
    }
}
